"""API服务器，为前端提供服务"""
from contextlib import asynccontextmanager
from fastapi import FastAPI, Header, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, List, Dict
import json
import contextvars
import logging
import re

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# 创建logger实例
logger = logging.getLogger(__name__)

# 移除本地 agent 导入，改为从 manager 导入
from module import agent_manager
from tools.data_query import set_global_token
from routes import kb_router, development_plan_router, model_config_router
from routes.agent_config import router as agent_config_router
from routes.tool_config import router as tool_config_router
from database import init_db
from module.chat_history import (
    get_history,
    add_message,
    DEFAULT_SESSION_ID
)
from config.config import (
    DEFAULT_MODEL,
    SERVER_PORT,
    SERVER_HOST,
)
from module.tts import RealtimeTTS
from module.asr import RealtimeASR
import asyncio
import base64

# 创建上下文变量，用于在工具中传递用户 token
request_token = contextvars.ContextVar('request_token', default=None)


def clean_markdown_for_tts(text: str) -> str:
    """清理Markdown格式,只保留纯文本用于TTS
    
    Args:
        text: 包含Markdown格式的文本
        
    Returns:
        清理后的纯文本
    """
    if not text:
        return ""
    
    # 移除Emoji表情符号（包括Email表情等）
    # 使用更精确的Emoji匹配模式，避免误删正常字符
    emoji_pattern = re.compile(
        "["
        "\U0001F600-\U0001F64F"  # 表情符号 😀-🙏
        "\U0001F300-\U0001F5FF"  # 符号和象形文字 🌀-🗿
        "\U0001F680-\U0001F6FF"  # 交通和地图符号 🚀-🛿
        "\U0001F1E0-\U0001F1FF"  # 旗帜 🇦-🇿
        "\U0001F900-\U0001F9FF"  # 补充符号和象形文字 🤀-🧿
        "\U0001FA00-\U0001FA6F"  # 扩展符号A 🨀-🩯
        "\U0001FA70-\U0001FAFF"  # 扩展符号B 🩰-🫿
        "\U00002600-\U000026FF"  # 杂项符号 ☀-⛿
        "\U00002700-\U000027BF"  # 装饰符号 ✀-➿
        "]+",
        flags=re.UNICODE
    )
    text = emoji_pattern.sub(' ', text)  # 替换为空格而不是删除，避免单词粘连
    
    # 移除代码块
    text = re.sub(r'```[\s\S]*?```', '', text)
    text = re.sub(r'`[^`]+`', '', text)
    
    # 移除链接,保留链接文本
    text = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', text)
    
    # 移除图片
    text = re.sub(r'!\[([^\]]*)\]\([^\)]+\)', '', text)
    
    # 移除标题标记
    text = re.sub(r'^#{1,6}\s+', '', text, flags=re.MULTILINE)
    
    # 移除粗体和斜体标记
    text = re.sub(r'\*\*([^\*]+)\*\*', r'\1', text)
    text = re.sub(r'\*([^\*]+)\*', r'\1', text)
    text = re.sub(r'__([^_]+)__', r'\1', text)
    text = re.sub(r'_([^_]+)_', r'\1', text)
    
    # 移除删除线
    text = re.sub(r'~~([^~]+)~~', r'\1', text)
    
    # 移除引用标记
    text = re.sub(r'^>\s+', '', text, flags=re.MULTILINE)
    
    # 移除列表标记
    text = re.sub(r'^\s*[-*+]\s+', '', text, flags=re.MULTILINE)
    text = re.sub(r'^\s*\d+\.\s+', '', text, flags=re.MULTILINE)
    
    # 移除水平线
    text = re.sub(r'^[-*_]{3,}$', '', text, flags=re.MULTILINE)
    
    # 移除HTML标签
    text = re.sub(r'<[^>]+>', '', text)
    
    # 移除多余的空行
    text = re.sub(r'\n{3,}', '\n\n', text)
    
    return text.strip()

SSE_EVENT_PREFIX = "__SSE_EVENT__"

SSE_EVENT_TYPE_MAPPING = {
    "time_series_data": "chart",
    "knowledge_search": "knowledge_search",
    "plan": "plan"
}

def process_sse_events(captured_output):
    """统一处理所有SSE事件
    
    Args:
        captured_output: 捕获的标准输出内容
        
    Yields:
        {"event": "...", "data": "..."} 格式的事件字典
    """
    lines = captured_output.split('\n')
    for line in lines:
        if SSE_EVENT_PREFIX in line:
            try:
                json_start = line.find(SSE_EVENT_PREFIX) + len(SSE_EVENT_PREFIX)
                json_str = line[json_start:]
                event_payload = json.loads(json_str)
                
                event_type = event_payload.get("event_type")
                data = event_payload.get("data", {})
                
                frontend_event = SSE_EVENT_TYPE_MAPPING.get(event_type, event_type)
                
                if event_type == "time_series_data":
                    yield {
                        "event": frontend_event,
                        "data": json.dumps({"type": "trend_chart", "data": data})
                    }
                elif event_type == "knowledge_search":
                    yield {
                        "event": frontend_event,
                        "data": json.dumps({"type": "knowledge_search", "data": data})
                    }
                elif event_type == "plan":
                    yield {
                        "event": frontend_event,
                        "data": json.dumps({"type": "plan", "data": data})
                    }
            except (json.JSONDecodeError, KeyError):
                pass


@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    # 初始化数据库
    try:
        init_db()
        print("数据库初始化完成")
    except Exception as e:
        print(f"数据库初始化失败: {e}")

    # 启动时初始化
    agent_manager.init_agent()
    yield
    # 关闭时清理（如果需要）


# 创建FastAPI应用
app = FastAPI(title="Agent API Server", version="1.0.0", lifespan=lifespan)

# 配置CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应该设置具体的域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(kb_router)
app.include_router(development_plan_router)
app.include_router(agent_config_router)
app.include_router(model_config_router)
app.include_router(tool_config_router)

class ChatRequest(BaseModel):
    """聊天请求模型"""
    message: str
    model: Optional[str] = None
    stream: Optional[bool] = True
    session_id: Optional[str] = DEFAULT_SESSION_ID
    checkpoint_id: Optional[str] = None  # 用于回滚的时间旅行检查点ID
    history: Optional[List[Dict]] = None  # 前端传来的历史记录（可选）
    token: Optional[str] = None  # 用户 token（可以从请求头或请求体获取）


class ChatResponse(BaseModel):
    """聊天响应模型"""
    content: str
    finished: bool = False


@app.get("/")
async def root():
    """根路径"""
    return {"message": "Agent API Server", "status": "running"}


@app.get("/health")
async def health():
    """健康检查"""
    agent = agent_manager.get_agent()
    return {"status": "healthy", "agent_initialized": agent is not None}


@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest, authorization: Optional[str] = Header(None)):
    """聊天接口（非流式）"""
    agent = agent_manager.get_agent()
    
    if not request.message.strip():
        return ChatResponse(content="消息不能为空", finished=True)
    
    # 提取 token：优先从请求头获取，否则从请求体获取
    token = None
    if authorization:
        # 从 Authorization 请求头提取 token (格式: Bearer {token})
        token = authorization.replace("Bearer ", "") if authorization.startswith("Bearer ") else authorization
    elif request.token:
        token = request.token
    
    # 同时设置到上下文变量和全局变量，确保工具能访问
    token_var = request_token.set(token)
    if token:
        set_global_token(token)  # 设置全局 token 作为备用方案
        print(f"[api] 🔐 Token 已设置到上下文和全局变量")
    
    try:
        from langchain_core.messages import HumanMessage, AIMessage
        
        # 获取会话ID
        session_id = request.session_id or DEFAULT_SESSION_ID
        
        # 构建历史消息列表
        history_messages = []
        
        # 如果前端提供了历史记录，使用前端的
        if request.history:
            for msg in request.history:
                content = msg.get("content", "").strip()
                if not content:
                    continue
                if msg.get("role") == "user":
                    history_messages.append(HumanMessage(content=content))
                elif msg.get("role") == "assistant":
                    history_messages.append(AIMessage(content=content))
        else:
            # 否则从后端存储获取历史记录
            history = get_history(session_id, max_messages=10)
            for msg in history:
                content = msg.get("content", "").strip()
                if not content:
                    continue
                if msg.get("role") == "user":
                    history_messages.append(HumanMessage(content=content))
                elif msg.get("role") == "assistant":
                    history_messages.append(AIMessage(content=content))
        
        # 添加当前用户消息
        history_messages.append(HumanMessage(content=request.message))
        
        # 收集完整响应
        full_response = ""
        
        # 流式调用agent
        for chunk in agent.stream({
            "messages": history_messages
        }):
            # 处理流式输出
            if "messages" in chunk:
                messages = chunk["messages"]
                if messages:
                    # 查找AI消息
                    for msg in messages:
                        if isinstance(msg, AIMessage) and hasattr(msg, 'content') and msg.content:
                            content = str(msg.content)
                            if content and content != full_response:
                                full_response = content
        
        return ChatResponse(content=full_response, finished=True)
    
    except Exception as e:
        import traceback
        error_detail = traceback.format_exc()
        logging.error(f"聊天错误: {error_detail}")
        return ChatResponse(content=f"错误: {str(e)}", finished=True)
    
    finally:
        # 重置 token 上下文和全局变量
        if token_var:
            request_token.reset(token_var)
        set_global_token(None)  # 清除全局 token


def generate_stream(request: ChatRequest, token: str = None):
    """生成流式响应
    
    Args:
        request: 聊天请求
        token: 用户 token（用于工具调用时的认证）
    """
    import io
    import sys
    
    agent = agent_manager.get_agent()
    agent_type = agent_manager.get_agent_type()
    
    if not request.message.strip():
        yield {
            "event": "message",
            "data": json.dumps({"content": "消息不能为空", "finished": True})
        }
        return
    
    # 同时设置 token 到上下文和全局变量
    # 这样无论通过哪种机制调用工具，都能获取到 token
    if token:
        token_var = request_token.set(token)
        set_global_token(token)  # 设置全局 token 作为备用方案
        print(f"[api] 🔐 Token 已设置到上下文和全局变量")
    else:
        token_var = None
        set_global_token(None)
    
    try:
        from langchain_core.messages import HumanMessage, AIMessage
        
        # 获取会话ID
        session_id = request.session_id or DEFAULT_SESSION_ID
        
        # 构建历史消息列表
        history_messages = []
        
        # 如果前端提供了历史记录，使用前端的（优先）
        if request.history:
            for msg in request.history:
                content = msg.get("content", "").strip()
                if not content:
                    continue
                if msg.get("role") == "user":
                    history_messages.append(HumanMessage(content=content))
                elif msg.get("role") == "assistant":
                    history_messages.append(AIMessage(content=content))
        else:
            # 否则从后端存储获取历史记录
            history = get_history(session_id, max_messages=10)  # 最多使用最近10条消息
            for msg in history:
                content = msg.get("content", "").strip()
                if not content:
                    continue
                if msg.get("role") == "user":
                    history_messages.append(HumanMessage(content=content))
                elif msg.get("role") == "assistant":
                    history_messages.append(AIMessage(content=content))
        
        # 添加当前用户消息
        current_user_message = HumanMessage(content=request.message)
        history_messages.append(current_user_message)
        
        # 保存用户消息到历史记录
        add_message(session_id, "user", request.message)
        
        # 构建配置
        config = {"configurable": {"thread_id": session_id}}
        if request.checkpoint_id:
            config["configurable"]["checkpoint_id"] = request.checkpoint_id
            print(f"[api] 🕰️ 使用时间旅行，回滚到检查点: {request.checkpoint_id}")
        
        # 设置输出重定向以捕获工具的 print 输出（用于发送趋势图数据等）
        old_stdout = sys.stdout
        output_capture = io.StringIO()
        sys.stdout = output_capture
        
        # 调用 agent 处理请求 - 实时发送消息到前端
        from langchain_core.messages import ToolMessage
        
        # 单一的 agent stream 调用，避免重复的 LLM 调用导致 tool_id 不匹配
        all_response_text = ""
        tool_calls_map = {}
        tool_calls_sent = set()
        tool_results_sent = {}
        
        # 使用多模式流式传输：同时获取更新（updates）和 LLM 令牌（messages）
        try:
            if agent_type == "deepagent":
                # DeepAgent 使用不同的调用方式
                result = agent.invoke({"messages": history_messages})
                if "messages" in result:
                    for msg in result["messages"]:
                        if hasattr(msg, 'content') and msg.content:
                            all_response_text = str(msg.content)
                            yield {
                                "event": "message",
                                "data": json.dumps({
                                    "content": all_response_text,
                                    "finished": False
                                })
                            }
            else:
                # LangGraph Agent 的流式处理
                for stream_mode, chunk in agent.stream(
                    {"messages": history_messages},
                    config=config,
                    stream_mode=["updates", "messages"],
                ):
                    if stream_mode == "messages":
                        # 处理 LLM 令牌流（token-level 流式输出）
                        message_chunk, metadata = chunk
                    
                        # 检查是否有内容
                        if hasattr(message_chunk, 'content') and message_chunk.content:
                            token_content = str(message_chunk.content)
                            if token_content:
                                all_response_text += token_content
                                
                                # 实时发送令牌到前端
                                try:
                                    yield {
                                        "event": "message",
                                        "data": json.dumps({
                                            "content": token_content,
                                            "finished": False
                                        })
                                    }
                                except GeneratorExit:
                                    print("[api] ⚠️ 客户端断开连接，停止生成")
                                    raise
                                except Exception as e:
                                    print(f"[api] ⚠️ 发送数据时出错: {e}")
                                    raise
                
                    elif stream_mode == "updates":
                        # 处理状态更新（用于工具调用和其他信息）
                        for node_name, node_output in chunk.items():
                            # 处理 agent 节点的输出
                            if node_name == "agent" and isinstance(node_output, dict) and "messages" in node_output:
                                messages = node_output["messages"]
                                for msg in messages:
                                    # 处理工具调用（在消息之后）
                                    if isinstance(msg, AIMessage) and hasattr(msg, 'tool_calls') and msg.tool_calls:
                                        for tool_call in msg.tool_calls:
                                            # 解析 tool_call
                                            if isinstance(tool_call, dict):
                                                tool_name = tool_call.get("name", "")
                                                tool_args = tool_call.get("args", {})
                                                tool_id = tool_call.get("id", "")
                                            else:
                                                tool_name = getattr(tool_call, "name", "")
                                                tool_args = getattr(tool_call, "args", {})
                                                tool_id = getattr(tool_call, "id", "")
                                            
                                            tool_id_str = str(tool_id) if tool_id else ""
                                            
                                            # 保存工具调用信息（用于后续匹配）
                                            if tool_id_str and tool_id_str not in tool_calls_sent:
                                                tool_calls_sent.add(tool_id_str)
                                                tool_calls_map[tool_id_str] = {
                                                    "name": tool_name,
                                                    "args": tool_args
                                                }
                                                
                                                # 实时发送工具调用开始事件
                                                try:
                                                    yield {
                                                        "event": "tool_call",
                                                        "data": json.dumps({
                                                            "type": "start",
                                                            "tool_name": tool_name,
                                                            "tool_args": tool_args,
                                                            "tool_id": tool_id
                                                        })
                                                    }
                                                except GeneratorExit:
                                                    # 客户端断开连接,停止生成
                                                    print("[api] ⚠️ 客户端断开连接，停止生成")
                                                    raise
                            
                            # 处理工具节点的输出
                            elif node_name == "tools" and isinstance(node_output, dict) and "messages" in node_output:
                                messages = node_output["messages"]
                                for msg in messages:
                                    if isinstance(msg, ToolMessage):
                                        tool_call_id = getattr(msg, 'tool_call_id', '')
                                        tool_result = str(msg.content) if hasattr(msg, 'content') else ""
                                        
                                        # 只发送一次每个工具的结果
                                        if tool_call_id and tool_call_id not in tool_results_sent:
                                            tool_results_sent[tool_call_id] = True
                                            
                                            # 找到对应的工具名称
                                            tool_name = ""
                                            tool_call_id_str = str(tool_call_id) if tool_call_id else ""
                                            
                                            if tool_call_id_str in tool_calls_map:
                                                tool_name = tool_calls_map[tool_call_id_str]["name"]
                                            
                                            # 实时发送工具执行结果
                                            try:
                                                yield {
                                                    "event": "tool_call",
                                                    "data": json.dumps({
                                                        "type": "result",
                                                        "tool_name": tool_name,
                                                        "tool_id": tool_call_id,
                                                        "result": tool_result
                                                    })
                                                }
                                            except GeneratorExit:
                                                # 客户端断开连接，停止生成
                                                print("[api] ⚠️ 客户端断开连接，停止生成")
                                                raise
                            
                            # 统一处理所有SSE事件
                            captured_output = output_capture.getvalue()
                            if SSE_EVENT_PREFIX in captured_output:
                                for event_item in process_sse_events(captured_output):
                                    try:
                                        yield event_item
                                    except GeneratorExit:
                                        print("[api] ⚠️ 客户端断开连接，停止生成")
                                        raise
                                # 清空已处理的输出
                                output_capture.truncate(0)
                                output_capture.seek(0)
        except GeneratorExit:
            # 客户端断开连接，正常退出
            print("[api] ⚠️ 客户端断开连接，停止生成")
            raise
        
        # 恢复标准输出
        sys.stdout = old_stdout
        
        # 获取当前状态的 Checkpoint ID
        new_checkpoint_id = None
        if hasattr(agent, "checkpointer") and agent.checkpointer:
             try:
                 # 获取最新状态
                 current_state = agent.get_state(config)
                 if current_state and current_state.config:
                     new_checkpoint_id = current_state.config["configurable"].get("checkpoint_id")
                     print(f"[api] 💾 新的 Checkpoint ID: {new_checkpoint_id}")
             except Exception as e:
                 print(f"[api] ⚠️ 获取 Checkpoint ID 失败: {e}")
        
        # 保存AI回复到历史记录
        ai_response_content = all_response_text if all_response_text else ""
        
        if ai_response_content:
            add_message(session_id, "assistant", ai_response_content)
        
        # 发送 Checkpoint ID 事件
        if new_checkpoint_id:
             yield {
                "event": "checkpoint",
                "data": json.dumps({
                    "checkpoint_id": new_checkpoint_id
                })
             }

        # 发送完成信号
        yield {
            "event": "message",
            "data": json.dumps({
                "content": "",
                "finished": True
            })
        }
    
    except GeneratorExit:
        # 客户端断开连接，正常退出
        print("[api] ⚠️ 客户端断开连接，停止生成")
        raise
    except Exception as e:
        import traceback
        error_msg = f"错误: {str(e)}\n{traceback.format_exc()}"
        try:
            yield {
                "event": "error",
                "data": json.dumps({
                    "content": error_msg,
                    "finished": True
                })
            }
        except GeneratorExit:
            # 客户端断开连接，正常退出
            print("[api] ⚠️ 客户端断开连接，停止生成")
            raise
    finally:
        # 恢复标准输出（防止意外残留）
        try:
            sys.stdout = old_stdout
        except:
            pass
        
        # 注意：generate_stream 作为生成器在线程池中执行，无法在不同 Context 中 reset contextvars
        # 因此仅清除全局变量，contextvars 会在请求完成后自动清理
        set_global_token(None)  # 清除全局 token


@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest, authorization: Optional[str] = Header(None)):
    """聊天接口（流式）"""
    from starlette.responses import StreamingResponse
    
    # 提取 token：优先从请求头获取，否则从请求体获取
    token = None
    if authorization:
        # 从 Authorization 请求头提取 token (格式: Bearer {token})
        token = authorization.replace("Bearer ", "") if authorization.startswith("Bearer ") else authorization
    elif request.token:
        token = request.token
    
    # 同时设置到上下文变量和全局变量，确保流式处理中能访问
    token_var = request_token.set(token)
    if token:
        set_global_token(token)  # 设置全局 token 作为备用方案
        print(f"[api] 🔐 Token 已设置到上下文和全局变量（流式）")
    
    def generate_sse_stream():
        """生成正确格式的 SSE 流"""
        try:
            for item in generate_stream(request, token):
                # item 应该是 {"event": "xxx", "data": "xxx"} 的格式
                if isinstance(item, dict):
                    event = item.get("event", "message")
                    data = item.get("data", "")
                    # 正确的 SSE 格式：完整事件块
                    sse_message = f"event: {event}\ndata: {data}\n\n"
                    yield sse_message.encode('utf-8')
        finally:
            # 注意：StreamingResponse 在线程池中执行生成器，无法在不同的 Context 中 reset contextvars
            # 因此仅清除全局变量，contextvars 会在请求完成后自动清理
            set_global_token(None)  # 清除全局 token
    
    return StreamingResponse(
        generate_sse_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"
        }
    )


class TTSRequest(BaseModel):
    """TTS请求模型"""
    text: str
    voice: Optional[str] = 'Cherry'
    language_type: Optional[str] = 'Chinese'
    sample_rate: Optional[int] = 24000
    format: Optional[str] = 'pcm'


class ChatWithTTSRequest(BaseModel):
    """带TTS的聊天请求模型"""
    message: str
    model: Optional[str] = None
    session_id: Optional[str] = DEFAULT_SESSION_ID
    checkpoint_id: Optional[str] = None
    history: Optional[List[Dict]] = None
    token: Optional[str] = None
    # TTS配置
    enable_tts: bool = True
    voice: Optional[str] = 'Cherry'
    language_type: Optional[str] = 'Chinese'
    sample_rate: Optional[int] = 24000
    

@app.post("/api/chat/stream-with-tts")
async def chat_stream_with_tts(request: ChatWithTTSRequest, authorization: Optional[str] = Header(None)):
    """
    聊天接口(流式) + 实时语音合成
    
    同时返回文本流和音频流
    """
    from starlette.responses import StreamingResponse
    
    # 提取 token
    token = None
    if authorization:
        token = authorization.replace("Bearer ", "") if authorization.startswith("Bearer ") else authorization
    elif request.token:
        token = request.token
    
    # 设置token到上下文
    token_var = request_token.set(token)
    if token:
        set_global_token(token)
        print(f"[api] 🔐 Token 已设置(TTS流式)")
    
    async def generate_chat_and_tts_stream():
        """生成聊天文本流和TTS音频流"""
        tts_client = None
        audio_chunks = []  # 使用列表而不是异步队列
        tts_task = None
        
        try:
            # 如果启用TTS,初始化TTS客户端
            if request.enable_tts:
                def audio_callback(audio_b64: str):
                    """TTS音频回调,将音频数据添加到列表"""
                    audio_chunks.append(audio_b64)
                
                tts_client = RealtimeTTS(
                    voice=request.voice,
                    language_type=request.language_type,
                    sample_rate=request.sample_rate
                )
                
                # 在线程池中建立TTS连接
                await asyncio.get_event_loop().run_in_executor(
                    None,
                    tts_client.connect,
                    audio_callback
                )
            
            # 创建聊天请求
            chat_req = ChatRequest(
                message=request.message,
                model=request.model,
                stream=True,
                session_id=request.session_id,
                checkpoint_id=request.checkpoint_id,
                history=request.history,
                token=token
            )
            
            # 用于累积文本块,批量发送给TTS
            text_buffer = ""
            buffer_size = 5  # 每累积10个字符发送一次
            last_audio_index = 0  # 跟踪已发送的音频索引
            
            # 流式获取聊天响应
            for item in generate_stream(chat_req, token):
                # 先发送已生成的音频块(在处理文本之前)
                if request.enable_tts and len(audio_chunks) > last_audio_index:
                    while last_audio_index < len(audio_chunks):
                        audio_b64 = audio_chunks[last_audio_index]
                        last_audio_index += 1
                        audio_event = f"event: audio\ndata: {json.dumps({'audio': audio_b64})}\n\n"
                        yield audio_event.encode('utf-8')
                
                if isinstance(item, dict):
                    event = item.get("event", "message")
                    data = item.get("data", "")
                    
                    # 发送文本消息到前端
                    sse_message = f"event: {event}\ndata: {data}\n\n"
                    yield sse_message.encode('utf-8')
                    
                    # 如果是文本消息且启用了TTS,发送给TTS合成
                    if event == "message" and request.enable_tts and tts_client:
                        try:
                            data_obj = json.loads(data)
                            content = data_obj.get("content", "")
                            finished = data_obj.get("finished", False)
                            
                            if content and not finished:
                                # 累积文本
                                text_buffer += content
                                
                                # 当缓冲区达到一定大小时,发送给TTS
                                if len(text_buffer) >= buffer_size:
                                    text_to_send = text_buffer
                                    text_buffer = ""
                                    
                                    # 清理Markdown格式
                                    clean_text = clean_markdown_for_tts(text_to_send)
                                    
                                    # 只有清理后还有内容才发送
                                    if clean_text.strip():
                                        # 在线程池中发送文本给TTS
                                        await asyncio.get_event_loop().run_in_executor(
                                            None,
                                            tts_client.synthesize_text,
                                            clean_text
                                        )
                                    
                                    # 等待一小段时间让TTS生成音频
                                    await asyncio.sleep(0.1)
                            
                            elif finished:
                                # 聊天完成,发送剩余文本给TTS
                                if text_buffer:
                                    # 清理Markdown格式
                                    clean_text = clean_markdown_for_tts(text_buffer)
                                    
                                    # 只有清理后还有内容才发送
                                    if clean_text.strip():
                                        await asyncio.get_event_loop().run_in_executor(
                                            None,
                                            tts_client.synthesize_text,
                                            clean_text
                                        )
                                    text_buffer = ""
                                
                                # 通知TTS结束
                                if tts_client:
                                    await asyncio.get_event_loop().run_in_executor(
                                        None,
                                        tts_client.finish
                                    )
                                    
                                    # 智能等待TTS完成:监控音频块数量变化,同时发送音频
                                    max_wait_time = 30.0  # 最大等待时间30秒
                                    check_interval = 0.2  # 每0.2秒检查一次
                                    stable_count = 0  # 稳定计数
                                    stable_threshold = 3  # 连续3次不变则认为完成
                                    last_chunk_count = len(audio_chunks)
                                    elapsed_time = 0.0
                                    
                                    while elapsed_time < max_wait_time:
                                        await asyncio.sleep(check_interval)
                                        elapsed_time += check_interval
                                        
                                        # 发送新生成的音频块
                                        while last_audio_index < len(audio_chunks):
                                            audio_b64 = audio_chunks[last_audio_index]
                                            last_audio_index += 1
                                            audio_event = f"event: audio\ndata: {json.dumps({'audio': audio_b64})}\n\n"
                                            yield audio_event.encode('utf-8')
                                        
                                        current_count = len(audio_chunks)
                                        if current_count > last_chunk_count:
                                            # 有新的音频块,重置稳定计数
                                            stable_count = 0
                                            last_chunk_count = current_count
                                        else:
                                            # 没有新音频块,增加稳定计数
                                            stable_count += 1
                                            if stable_count >= stable_threshold:
                                                # 连续多次检查都没有新音频,认为完成
                                                break
                                        
                        except json.JSONDecodeError:
                            pass
            
            # 发送剩余的音频块
            if request.enable_tts:
                # 再等待一小段时间,确保所有音频回调都已执行
                await asyncio.sleep(0.2)
                
                while last_audio_index < len(audio_chunks):
                    audio_b64 = audio_chunks[last_audio_index]
                    last_audio_index += 1
                    audio_event = f"event: audio\ndata: {json.dumps({'audio': audio_b64})}\n\n"
                    yield audio_event.encode('utf-8')
                
                # 发送音频结束标记
                audio_done = f"event: audio_done\ndata: {json.dumps({'finished': True})}\n\n"
                yield audio_done.encode('utf-8')
                        
        except Exception as e:
            import traceback
            error_msg = f"错误: {str(e)}\n{traceback.format_exc()}"
            logging.error(f"聊天+TTS错误: {error_msg}")
            error_event = f"event: error\ndata: {json.dumps({'content': error_msg})}\n\n"
            yield error_event.encode('utf-8')
            
        finally:
            # 清理TTS客户端
            if tts_client:
                try:
                    await asyncio.get_event_loop().run_in_executor(
                        None,
                        tts_client.close
                    )
                except:
                    pass
            
            # 清理token
            set_global_token(None)
    
    return StreamingResponse(
        generate_chat_and_tts_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"
        }
    )


@app.websocket("/ws/asr")
async def websocket_asr(websocket: WebSocket):
    """WebSocket端点：实时语音识别
    
    前端通过WebSocket发送音频数据，后端实时返回识别结果
    """
    await websocket.accept()
    logger.info("ASR WebSocket连接已建立")
    
    asr_client = None
    
    try:
        # 接收配置消息
        config_msg = await websocket.receive_json()
        
        # 提取配置参数
        language = config_msg.get('language', 'zh')
        sample_rate = config_msg.get('sample_rate', 16000)
        input_audio_format = config_msg.get('input_audio_format', 'pcm')
        corpus_text = config_msg.get('corpus_text', None)
        enable_turn_detection = config_msg.get('enable_turn_detection', True)
        
        logger.info(f"ASR配置: language={language}, sample_rate={sample_rate}, format={input_audio_format}")
        
        # 定义回调函数
        async def on_transcription(text: str, is_complete: bool):
            """识别结果回调"""
            try:
                await websocket.send_json({
                    'type': 'transcription',
                    'text': text,
                    'is_complete': is_complete
                })
            except Exception as e:
                logger.error(f"发送识别结果失败: {e}")
        
        async def on_speech_start():
            """语音开始回调"""
            try:
                await websocket.send_json({
                    'type': 'speech_start'
                })
            except Exception as e:
                logger.error(f"发送语音开始事件失败: {e}")
        
        async def on_speech_stop():
            """语音停止回调"""
            try:
                await websocket.send_json({
                    'type': 'speech_stop'
                })
            except Exception as e:
                logger.error(f"发送语音停止事件失败: {e}")
        
        async def on_error(error_msg: str):
            """错误回调"""
            try:
                await websocket.send_json({
                    'type': 'error',
                    'message': error_msg
                })
            except Exception as e:
                logger.error(f"发送错误消息失败: {e}")
        
        # 创建ASR客户端
        asr_client = RealtimeASR(
            language=language,
            sample_rate=sample_rate,
            input_audio_format=input_audio_format,
            corpus_text=corpus_text,
            enable_turn_detection=enable_turn_detection
        )
        
        # 获取事件循环
        loop = asyncio.get_event_loop()
        
        # 在线程池中建立ASR连接（因为是同步操作）
        def sync_connect():
            asr_client.connect(
                on_transcription=lambda text, is_complete: asyncio.run_coroutine_threadsafe(
                    on_transcription(text, is_complete), loop
                ),
                on_speech_start=lambda: asyncio.run_coroutine_threadsafe(
                    on_speech_start(), loop
                ),
                on_speech_stop=lambda: asyncio.run_coroutine_threadsafe(
                    on_speech_stop(), loop
                ),
                on_error=lambda msg: asyncio.run_coroutine_threadsafe(
                    on_error(msg), loop
                )
            )
        
        await loop.run_in_executor(None, sync_connect)
        
        # 发送连接成功消息
        await websocket.send_json({
            'type': 'connected',
            'message': 'ASR连接已建立'
        })
        
        # 接收音频数据并发送给ASR
        while True:
            try:
                message = await websocket.receive()
                
                # 处理文本消息（JSON格式）
                if 'text' in message:
                    data = json.loads(message['text'])
                    msg_type = data.get('type', '')
                    
                    if msg_type == 'audio':
                        # 音频数据（Base64编码）
                        audio_b64 = data.get('audio', '')
                        if audio_b64 and asr_client:
                            # 在线程池中发送音频
                            await asyncio.get_event_loop().run_in_executor(
                                None,
                                asr_client.append_audio,
                                audio_b64
                            )
                    
                    elif msg_type == 'commit':
                        # 手动提交（仅在禁用VAD时使用）
                        if asr_client:
                            await asyncio.get_event_loop().run_in_executor(
                                None,
                                asr_client.commit
                            )
                    
                    elif msg_type == 'close':
                        # 客户端请求关闭
                        logger.info("收到客户端关闭请求")
                        break
                
                # 处理二进制消息（原始音频数据）
                elif 'bytes' in message:
                    # 将二进制数据转为Base64
                    audio_bytes = message['bytes']
                    audio_b64 = base64.b64encode(audio_bytes).decode('utf-8')
                    if asr_client:
                        await asyncio.get_event_loop().run_in_executor(
                            None,
                            asr_client.append_audio,
                            audio_b64
                        )
            
            except WebSocketDisconnect:
                logger.info("客户端断开连接")
                break
            except Exception as e:
                logger.error(f"处理消息失败: {e}")
                await on_error(f"处理消息失败: {str(e)}")
    
    except WebSocketDisconnect:
        logger.info("WebSocket连接已断开")
    except Exception as e:
        import traceback
        error_detail = traceback.format_exc()
        logger.error(f"ASR WebSocket错误: {e}\n{error_detail}")
        try:
            await websocket.send_json({
                'type': 'error',
                'message': str(e)
            })
        except:
            pass
    finally:
        # 关闭ASR连接
        if asr_client:
            try:
                await asyncio.get_event_loop().run_in_executor(
                    None,
                    asr_client.close
                )
            except Exception as e:
                logger.error(f"关闭ASR连接失败: {e}")
        
        # 关闭WebSocket
        try:
            await websocket.close()
        except:
            pass


if __name__ == "__main__":
    import uvicorn
    print(f"🚀 启动服务器: http://{SERVER_HOST}:{SERVER_PORT}")
    uvicorn.run(app, host=SERVER_HOST, port=SERVER_PORT)

